package com.piece.core.rabbit.config;

import com.piece.core.framework.constant.RegisterConstants;
import com.piece.core.framework.support.convert.Convert;
import com.piece.core.framework.util.message.MessageFactory;
import com.piece.core.rabbit.consumer.RabbitMessageConsumer;
import com.piece.core.rabbit.properties.RabbitProperties;
import com.piece.core.rabbit.producer.RabbitMessageProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor;
import org.springframework.amqp.rabbit.config.RabbitListenerConfigUtils;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Lazy;
import org.springframework.context.annotation.Role;
import java.util.HashMap;
import java.util.Map;

/**
 * rabbitmq 配置类
 */
@Slf4j
public class RabbitAutoConfig {

    @Autowired
    private RabbitProperties rabbitProperties;

    @Bean(name = "connectionFactory")
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setVirtualHost(rabbitProperties.getVirtualHost());
        factory.setHost(rabbitProperties.getHost());
        factory.setPort(rabbitProperties.getPort());
        factory.setUsername(rabbitProperties.getUsername());
        factory.setPassword(rabbitProperties.getPassword());
        factory.setConnectionTimeout(Convert.toInt(rabbitProperties.getConnectionTimeout().getSeconds(), 60000));

        // 设置发布消息后回调
        factory.setPublisherReturns(true);
        // 设置发布后确认类型
        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        factory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        return factory;
    }

    /**
     * 单一消费者
     */
    @Bean(name = "singleListenerContainer")
    public RabbitListenerContainerFactory singleListenerContainer(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setPrefetchCount(rabbitProperties.getListener().getSimple().getPrefetch());
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(rabbitProperties.getListener().getSimple().getAcknowledgeMode());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        return factory;
    }

    /**
     * 多个消费者
     */
    @Bean(name = "multiListenerContainer")
    public RabbitListenerContainerFactory multiListenerContainer(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setPrefetchCount(rabbitProperties.getListener().getSimple().getPrefetch());
        factory.setConcurrentConsumers(rabbitProperties.getListener().getSimple().getConcurrency());
        factory.setMaxConcurrentConsumers(rabbitProperties.getListener().getSimple().getMaxConcurrency());
        factory.setAcknowledgeMode(rabbitProperties.getListener().getSimple().getAcknowledgeMode());
        return factory;
    }

    /**
     * 延迟队列交换机
     */
    @Bean(name = "registerDelayExchange")
    public DirectExchange registerDelayExchange() {
        return new DirectExchange(RegisterConstants.DELAY_EXCHANGE_NAME, false, false);
    }

    /**
     * 延迟队列
     */
    @Bean(name = "registerDelayQueue")
    public Queue registerDelayQueue() {
        Map<String, Object> params = new HashMap<>();
        params.put("x-dead-letter-exchange", RegisterConstants.EXCHANGE_NAME);
        params.put("x-dead-letter-routing-key", RegisterConstants.ROUTE_NAME);
        return new Queue(RegisterConstants.DELAY_QUEUE_NAME, false, false, false, params);
    }

    /**
     * 绑定延迟队列和交换机
     */
    @Bean
    public Binding registerDelayBinding(DirectExchange registerDelayExchange, Queue registerDelayQueue) {
        return BindingBuilder
                .bind(registerDelayQueue)
                .to(registerDelayExchange)
                .with(RegisterConstants.DELAY_ROUTE_NAME);
    }

    /**
     * 主题队列交换机
     */
    @Bean(name = "registerTopicExchange")
    public TopicExchange registerTopicExchange() {
        return new TopicExchange(RegisterConstants.EXCHANGE_NAME, false, false);
    }

    /**
     * 主题队列
     */
    @Bean(name = "registerQueue")
    public Queue registerQueue() {
        return new Queue(RegisterConstants.QUEUE_NAME, false, false, false, null);
    }

    /**
     * 绑定主题队列和交换机
     */
    @Bean
    public Binding registerBinding(TopicExchange registerTopicExchange, Queue registerQueue) {
        return BindingBuilder
                .bind(registerQueue)
                .to(registerTopicExchange)
                .with(RegisterConstants.ROUTE_NAME);
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        // 服务启动时候开启自动启动
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);

        // TODO： 后续可完善消息确认逻辑
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (log.isDebugEnabled()) {
                if (ack) {
                    log.debug("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
                } else {
                    log.debug("消息发送失败:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
                }
            }
        });

        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            if (log.isDebugEnabled()) {
                log.debug("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message);
            }
        });

        return rabbitTemplate;
    }

    @Bean("rabbitMessageProducer")
    public RabbitMessageProducer rabbitMessageProducer(RabbitTemplate rabbitTemplate, @Lazy MessageFactory messageFactory) {
        return new RabbitMessageProducer(rabbitTemplate, messageFactory);
    }

    @Bean("rabbitMessageConsumer")
    public RabbitMessageConsumer rabbitMessageConsumer(@Lazy MessageFactory messageFactory) {
        return new RabbitMessageConsumer(messageFactory);
    }

    @Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public RabbitListenerAnnotationBeanPostProcessor defaultRabbitListenerAnnotationBeanPostProcessor() {
        return new RabbitListenerAnnotationBeanPostProcessor();
    }

    @Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
    public RabbitListenerEndpointRegistry defaultRabbitListenerEndpointRegistry() {
        return new RabbitListenerEndpointRegistry();
    }
}
